package net.neptech.listener;



import com.fasterxml.jackson.databind.ObjectMapper;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import net.neptech.constants.order.OrderTopics;
import net.neptech.model.appointment.Appointment;
import net.neptech.model.dto.AppointmentDTO;
import net.neptech.dto.order.DelayMessage;
import net.neptech.model.emum.AnnStatusEnum;
import net.neptech.service.IAppointmentService;
import net.neptech.topics.ClinicStaffTopics;
import net.neptech.utils.BeanCopyUtils;
import net.neptech.vm.ResponseResult;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.stereotype.Component;

/**
 * 延迟消息消费者
 * 处理从Kafka接收到的延迟消息
 *
 * @author neptech
 */
@Slf4j
@Component
public class DelayMessageConsumer {

    @Resource
    private ObjectMapper objectMapperKa;
    @Autowired
    private IAppointmentService appointmentService;

    @Autowired
    private RedisTemplate<Object,Object> redisTemplate;

    @Autowired
    private KafkaTemplate<String,Object> kafkaTemplate;


    /**
     * 消费延迟消息
     *
     * @param delayMessage 延迟消息
     * @param partition 分区
     * @param offset 偏移量
     * @param acknowledgment 确认对象
     */
    @KafkaListener(topics = "${delayed.queue.topic}", groupId = "${spring.kafka.consumer.group-id}")
    public void consumeDelayMessage(
            @Payload DelayMessage delayMessage,
            @Header(KafkaHeaders.RECEIVED_PARTITION) int partition,
            @Header(KafkaHeaders.OFFSET) long offset,
            Acknowledgment acknowledgment) {

        log.info("接收到延迟消息，messageId: {}, messageType: {}, partition: {}, offset: {}",
                delayMessage.getMessageId(), delayMessage.getMessageType(), partition, offset);

        try {
            // 根据消息类型处理不同的业务逻辑
            switch (delayMessage.getMessageType()) {
                case "ORDER_TIMEOUT":
                    handleOrderTimeout(delayMessage);
                    break;
                default:
                    log.warn("未知的消息类型: {}", delayMessage.getMessageType());
                    break;
            }

            // 手动确认消息
            acknowledgment.acknowledge();
            log.info("延迟消息处理完成，messageId: {}", delayMessage.getMessageId());

        } catch (Exception e) {
            log.error("处理延迟消息异常，messageId: {}, messageType: {}",
                    delayMessage.getMessageId(), delayMessage.getMessageType(), e);

            // 根据业务需求决定是否确认消息
            // 如果不确认，消息会重新投递
            // 这里选择确认消息，避免无限重试
            acknowledgment.acknowledge();
        }
    }

    /**
     * 处理订单超时消息
     *
     * @param delayMessage 延迟消息
     */
    private void handleOrderTimeout(DelayMessage delayMessage) {
        try {

            Appointment appointment = delayMessage.getContent();
            Long orderId = appointment.getOrderId();
            Byte status = appointment.getStatus();
            //先查数据库  再处理

            log.info("处理订单超时消息，订单号: {}, 状态: {} ", orderId,status);

            if (status==0) {
                // 执行订单超时处理 取消预约关闭订单
                //号放回 库存
                byte code = AnnStatusEnum.CANCELLED.getCode();
                log.info("订单超时，设置预约订单号: {}, 状态: {} ", orderId,code);

                ResponseResult<Void> result = appointmentService.updateAppointmentStatus(orderId, code);
//                orderService.handleOrderTimeout(orderNo);


                if (result.getCode() == 200){
                    //把号还回去
                    redisTemplate.opsForList().leftPush("xxyy:doctor:"+appointment.getStaffId()+":clnicId:"
                                    +appointment.getClnicId(),
                            appointment.getSlotNo());
                    kafkaTemplate.send(OrderTopics.ORDER_HANDLE_TOPIC,"deleteOrder"
                            , appointment.getOrderId());
                    //放回库存
                   ;
                    kafkaTemplate.send(ClinicStaffTopics.SLOT_USE_UPDATE,"++slot",
                            BeanCopyUtils.copyBean(appointment, AppointmentDTO.class));
                    //消息队列还订单
                }else {
                    log.error("订单超时，设置预约订单号: {}, 状态: {} ", orderId,code);
                }
                //重写投放账号
                log.info("超时处理");
            } else {
                log.warn("未知的订单超时动作 2 是未支付: {}", status);
            }

        } catch (Exception e) {
            log.error("处理订单超时消息异常，messageId: {}", delayMessage.getMessageId(), e);
            throw new RuntimeException("处理订单超时消息失败", e);
        }
    }
}
